#ifndef _CHUNKSTORE_H
#define _CHUNKSTORE_H
#include "dss_common.h"
#include "MiRpc.h"

#include "spdk/stdinc.h"
#include "spdk/bdev.h"
#include "spdk/env.h"
#include "spdk/event.h"
#include "spdk/blob_bdev.h"
#include "spdk/blob.h"
#include "spdk/log.h"
#include "spdk/string.h"
#include "spdk/bdev_module.h"

#include <unordered_map>
#include <vector>
#define MAX_CK_NUM 128
class ChunkServer
{
    uint16_t cs_id_ = -1; // chunkserver global id
    char *cs_conf_  = nullptr;

    void *mem_ = nullptr; //alloc by spdk_dma_zalloc() 
    uint32_t mem_size_ = 0;
    uint32_t chunk_size_ = 0;
    uint32_t heartbeat_interval_ = 0;
    uint32_t cs_num_ = 0;

    struct spdk_thread *thread_ = nullptr;
    struct spdk_poller *rpc_io_poller_ = nullptr;
    struct spdk_poller *rpc_conn_poller_ = nullptr;
    struct spdk_poller *blobstore_io_poller_ = nullptr;
    struct spdk_poller *heartbeat_poller_ = nullptr;
   
    int rc_ = 0;
    uint64_t last_ = 0; //last send heartbeat
    /*
    * spdk blobstore
    */
    char *bdev_json_file_ = nullptr;
    char *spdk_rpc_uri_ = nullptr;
    char bdev_name_[16]; //device name to create blobstore
    struct spdk_bs_dev *bs_bdev_ = nullptr;
    struct spdk_io_channel *io_channel_ = nullptr; //io channel to submit blob r/w request
    
    struct spdk_blob_store *bs_ = nullptr; // blobstore context
    std::unordered_map<uint64_t, struct spdk_blob*> id2chunk_; //map<blob_id, chunk>
    

    bool is_restart_ = false;

    int close_blob_num_ = 0;


    /*
    * minirpc
    */
    MemoryManager *mm_ = nullptr;
    char *local_uri_ = nullptr;
    char *master_uri_ = nullptr;
    char *cs_uri_[MAX_CK_NUM] = {nullptr};
    uint32_t master_sess_id_ = 0;
    uint32_t cs_sess_id_[MAX_CK_NUM];
    RdmaRpc *rpc_server_ = nullptr;
    RdmaRpc *rpc_client_ = nullptr;
     
public:
    uint64_t io_unit_size_ = 0; //min io size of blob r/w request  
    uint64_t total_cluster_num_ = 0;
    uint64_t free_cluster_num_ = 0;
    uint64_t cluster_size_ = 0;
    std::vector<uint64_t> blobs_ = {};
    std::vector<uint64_t>::size_type reuse_blob_ = 0;

    ChunkServer() = default;
    ChunkServer(const ChunkServer&) = delete;
    ChunkServer &operator=(const ChunkServer&) = delete; 
    ~ChunkServer() = default;
    uint64_t get_cluster_per_chunk() const
    {
        return chunk_size_/cluster_size_;
    }
    uint64_t io_unit_size() const
    {
        return io_unit_size_;
    }
    void set_cs_id(uint16_t cs_id)
    {
        cs_id_ = cs_id;
    }
    uint16_t cs_id() const
    {
        return cs_id_;
    }
    
    char **cs_uri()
    {
        return cs_uri_;
    }
    void set_cs_conf(char *cs_conf)
    {
        cs_conf_ = cs_conf;
    }
    const char *cs_conf()
    {
        return cs_conf_;
    }

    void set_is_restart(bool flag)
    {
        is_restart_ = true;
    }
    bool is_restart() const
    {
        return is_restart_;
    }
    
    const char *bdev_json_file() const
    {
        return bdev_json_file_;
    }

    const char *spdk_rpc_uri() const
    {
        return spdk_rpc_uri_;
    }

    void set_mem(void* mem)
    {
        mem_ = mem;
    }

    void* mem() const
    {
        return mem_;
    }

    void set_mem_size(uint32_t mem_size)
    {
        mem_size_ = mem_size;
    }

    uint32_t mem_size() const
    {
        return mem_size_;
    }

    void set_bdev_name(char* bdev_name)
    {
        strcpy(bdev_name_, bdev_name);
    }

    const char* bdev_name() const
    {
        return bdev_name_;
    }

    void set_bs_dev(struct spdk_bs_dev* bs_dev)
    {
        bs_bdev_ = bs_dev;
    }

    const struct spdk_bs_dev* bs_bdev() const
    {
        return bs_bdev_;
    }

    std::unordered_map<uint64_t, struct spdk_blob*> &id2chunk() 
    {
        return id2chunk_;
    }
    void set_rc(int rc)
    {
        rc_ = rc;
    }

    int rc()
    {
        return rc_;
    }

    void set_heartbeat_interval(uint32_t heartbeat_interval)
    {
        heartbeat_interval_ = heartbeat_interval;

    }

    const uint32_t heartbeat_interval()
    {
        return heartbeat_interval_;
    }


    int ParseConf(const char* conf);
    void FreeConf()
    {
        free(cs_conf_);
    }
    
    void set_thread(struct spdk_thread* thread)
    {
        thread_ = thread;
    }

    const struct spdk_thread* thread()
    {
        return thread_;
    }

    void set_rpc_server(RdmaRpc *rpc)
    {
        rpc_server_ = rpc; 
    }

    RdmaRpc* rpc_server() const
    {
        return rpc_server_;
    }

    void set_rpc_client(RdmaRpc *rpc)
    {
        rpc_client_ = rpc; 
    }

    RdmaRpc* rpc_client() const
    {
        return rpc_client_;
    }

    void set_io_channel(struct spdk_io_channel *io_channel)
    {
        io_channel_ = io_channel;
    }

    struct spdk_io_channel *io_channel() const
    {
        return io_channel_;
    } 
      
    void set_bs(struct spdk_blob_store *bs)
    {
        bs_ = bs;
    }

    struct spdk_blob_store *bs() const
    {
        return bs_;
    }

    MemoryManager* mm()
    {
        return mm_;
    }

    void set_mm(MemoryManager *mm)
    {
        mm_ = mm;
    }

    void parse_bs_inf()
    {
        io_unit_size_ = spdk_bs_get_io_unit_size(bs_);
        total_cluster_num_ = spdk_bs_total_data_cluster_count(bs_);
        free_cluster_num_ = spdk_bs_free_cluster_count(bs_);
        cluster_size_ = spdk_bs_get_cluster_size(bs_);
        SPDK_NOTICELOG("io_unit_size = %d\n", io_unit_size_);
        SPDK_NOTICELOG("total_cluster_num = %d\n", total_cluster_num_);
        SPDK_NOTICELOG("free_cluster_num = %d\n", free_cluster_num_);
        SPDK_NOTICELOG("cluster_size = %d\n", cluster_size_);
    }

    void prepare_heartbeat_msg(RpcMsg* msg)
    {
        struct dss_msg* dss_header = static_cast<struct dss_msg*>(rpc_client_->PrepareMsg(msg, M_HEARTBEAT_SERVICE));
		
		msg->data_len = sizeof(struct dss_msg) + sizeof(struct heartbeat_req);
		struct heartbeat_req* heart_header = (struct heartbeat_req*)dss_header->data;
		heart_header->cs_id = cs_id_;
		heart_header->io_unit_size = io_unit_size_;
		heart_header->cluster_total = total_cluster_num_;
		heart_header->cluster_free = free_cluster_num_;
		heart_header->cluster_size = cluster_size_;

		dss_header->client_id = -1;
		dss_header->msg_type = CS_HEARTBEAT;
		dss_header->msg_id = 0;
		dss_header->data_len = sizeof(struct heartbeat_req);

    }
    
    uint32_t master_sess_id() const
    {
        return master_sess_id_;
    }
    void set_close_blob_num(int val)
    {
        close_blob_num_ = val;
    }

    int close_blob_num()
    {
        return close_blob_num_;
    }

    void set_last(uint64_t last)
    {
        last_ = last;
    }

    const uint64_t last() const
    {
        return last_;
    }

    void RegisterRpcConnPoller();
    void UnRegisterRpcConnPoller()
    {
        spdk_poller_unregister(&rpc_conn_poller_);
    }
    
    void RegisterRpcIoPoller();
    void UnRegisterRpcIoPoller()
    {
        spdk_poller_unregister(&rpc_io_poller_);
		
    }

    
    void RegisterBlobstoreIoPoller();
    void UnRegisterBlobstoreIoPoller()
    {
        spdk_poller_unregister(&blobstore_io_poller_);
    }

    int ConnectToMaster();
    void RegisterHeartbeatPoller();
    void UnRegisterHeartbeatPoller();

    friend void cs_client_auth_service(void* self, void* ex, void* pro);
    friend void cs_object_io_service(void* self, void* ex, void* pro);
};

struct ChunkContext
{
    struct spdk_blob* blob; /**SPDK blobstore上下文**/
    enum IDENTITY{HEAD, HEAD_NEXT, MED, TAIL_PREV, TAIL} identity;  /**chunk在链上的位置**/
    std::unordered_map<uint32_t, struct chunk_op*> pending_map;  /**写请求pending集合，通过版本号查询**/
    uint16_t current_version;  /**给予下一个写请求的版本号**/
    uint16_t latest_version;  /**已经落盘的最新版本的写请求，当chunk非链尾的chunkserver处理读请求，且数据dirty(pending_map非空)，询问链尾的chunkserver最新落盘的数据将返回该值**/
};

/**ChunkServer Write Operation Context for Replicas**/  
struct WriteOp  
{  
    uint16_t version;    /**写请求版本号**/  
    bool is_sent;    /**是否已经转发给后继**/  
};  

/**ChunkServer Chunk Operation**/  
struct ChunkOp
{  
    RpcExecution *rpc_ex;    /**RPC请求上下文**/  
    RdmaRpc *rpc_self;    /**RPC实例**/  
    ChunkOp *object_ctx;    /**object上下文**/ 
    ChunkServer *cs_self;
    struct dss_msg* dss_request;    /**解析RPC请求得到的创、删、读、写请求**/  
    struct dss_msg* dss_response;    /**通过RPC应答返回的****应答**/  
    void *alloc_buffer;
    void *read_modify_write_buffer;
    WriteOp write_op_ctx;   /**链式复制写请求相关信息**/  
};  

#endif